package com.shujia.dws

import com.shujia.utils.SparkTool
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql._

object DwsScenicTouristMskDay extends SparkTool {
  override def run(spark: SparkSession): Unit = {
    import spark.implicits._
    import org.apache.spark.sql.functions._


    //1、读取融合表
    val mergeLocation: Dataset[Row] = spark
      .table("dwi.dwi_res_regn_mergelocation_msk_d")
      .where($"day_id" === day_id)


    //2|读取用户画像表
    val usertag: Dataset[Row] = spark
      .table("dim.dim_usertag_msk_d")
      .where($"day_id" === day_id)

    //景区配置表
    val scenicGrid: Dataset[Row] = spark
      .table("dim.dim_scenic_grid")
      //将网格列表拆分出来
      .select($"scenic_id", $"scenic_name", $"county_id", $"grids", explode(split($"grids", ",")) as "grid")

    /**
      * 景区游客判断条件
      * 1、常住地不在景区内
      * 2、目的地出现在景区内，停留时间大于30分钟
      *
      */

    val resultDF: DataFrame = mergeLocation
      //关联位置融合表和景区网格表，关联上了说明用户出现在景区内
      .join(scenicGrid.hint("broadcast"), $"grid_id" === $"grid", "inner")
      //计算用户在景区内的停留时间
      .withColumn("min_date", min($"start_date") over Window.partitionBy($"mdn", $"scenic_id"))
      .withColumn("max_date", max($"end_date") over Window.partitionBy($"mdn", $"scenic_id"))
      .withColumn("diff_date", unix_timestamp($"max_date", "yyyyMMddHHmmss") - unix_timestamp($"min_date", "yyyyMMddHHmmss"))
      //停留时间大于30分钟
      .where($"diff_date" > 1800)
      //关联获取用户常住地的网格
      .join(usertag, "mdn")
      //1、常住地不在景区内
      .where(array_contains(split($"grids", ","), $"resi_grid_id"))
      //整理数据
      .select($"mdn", $"resi_county_id", $"scenic_id", $"scenic_name", $"min_date", round($"diff_date" / 3600.0, 3))
      .distinct()

    //保存数据
    resultDF
      .write
      .format("csv")
      .option("sep", "\t")
      .mode(SaveMode.Overwrite)
      .save(s"/daas/motl/dws/dws_scenic_tourist_msk_d/day_id=$day_id")

    //增肌分区
    spark.sql(
      s"""
        |alter table  dws.dws_scenic_tourist_msk_d add if not exists partition(day_id='$day_id')
        |
      """.stripMargin)

  }
}
